package com.aiden.rxjavatest.activity;

import android.annotation.SuppressLint;
import android.nfc.Tag;
import android.support.v7.app.AppCompatActivity;
import android.os.Bundle;
import android.util.Log;
import android.util.TimeUtils;
import android.view.View;

import com.aiden.rxjavatest.R;

import java.io.Serializable;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Locale;
import java.util.Random;
import java.util.concurrent.Callable;
import java.util.concurrent.TimeUnit;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.ObservableSource;
import io.reactivex.Observer;
import io.reactivex.Single;
import io.reactivex.SingleObserver;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.BiFunction;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
import io.reactivex.functions.Predicate;
import io.reactivex.schedulers.Schedulers;

@SuppressWarnings("CheckResult")
public class MainActivity extends AppCompatActivity {

    private static final String TAG = "RxJava";

    private static final SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss", Locale.CHINA);

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);

        findViewById(R.id.btn_permission).setOnClickListener(new View.OnClickListener() {
            @Override
            public void onClick(View v) {

            }
        });

        create();

        map();

        zip();

        concat();

        flatMap();

        concatMap();

        distinct();

        filter();

        buffer();

        timer();

        interval();

        intervalRange();

        doOnNext();

        skip();

        take();

        single();

        debounce();

        defer();

        last();

        merge();

        reduce();

        scan();

        window();
    }

    @Override
    protected void onDestroy() {
        super.onDestroy();
        if (mIntervalDisposable != null && !mIntervalDisposable.isDisposed()) {
            mIntervalDisposable.dispose();
        }
    }

    private void create() {
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                Log.e(TAG, "Observable emitter 1\n");
                emitter.onNext(1);
                Log.e(TAG, "Observable emitter 2\n");
                emitter.onNext(2);
                Log.e(TAG, "Observable emitter 3\n");
                emitter.onNext(3);
                emitter.onComplete();
                Log.e(TAG, "Observable emitter 4\n");
                emitter.onNext(4);
            }
        }).subscribe(new Observer<Integer>() {
            private Disposable mDisposable;
            private int i;

            @Override
            public void onSubscribe(Disposable d) {
                mDisposable = d;
                Log.e(TAG, "onSubscribe: " + mDisposable.isDisposed() + "\n");
            }

            @Override
            public void onNext(Integer integer) {
                i++;
                Log.e(TAG, "onNext: value :" + integer + "\n");
                if (i == 2) {
                    mDisposable.dispose();
                    Log.e(TAG, "onNext: isDisposable : " + mDisposable.isDisposed() + "\n");
                }
            }

            @Override
            public void onError(Throwable e) {
                Log.e(TAG, "onError: message : " + e.getMessage());
            }

            @Override
            public void onComplete() {
                Log.e(TAG, "onComplete");
            }
        });
    }

    private void map() {
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
            }
        }).map(new Function<Integer, String>() {
            @Override
            public String apply(Integer integer) throws Exception {
                return "This is result " + integer;
            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                Log.e(TAG, "accept : " + s);
            }
        });
    }

    private void zip() {
        Observable.zip(getStringObservable(), getIntegerObservable(), new BiFunction<String, Integer, String>() {
            @Override
            public String apply(String s, Integer integer) throws Exception {
                return "zip result : " + s + integer;
            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) {
                Log.e(TAG, s);
            }
        });
    }

    private Observable<String> getStringObservable() {
        return Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                emitter.onNext("A");
                emitter.onNext("B");
                emitter.onNext("C");
                emitter.onNext("D");
                emitter.onComplete();
            }
        });
    }

    private Observable<Integer> getIntegerObservable() {
        return Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
                emitter.onNext(4);
                emitter.onNext(5);
                emitter.onComplete();
            }
        });
    }


    private void concat() {
        Observable.concat(getStringObservable(),
                getIntegerObservable().map(new Function<Integer, String>() {
                    @Override
                    public String apply(Integer integer) throws Exception {
                        return String.valueOf(integer);
                    }
                }))
                .subscribe(new Consumer<String>() {
                    @Override
                    public void accept(String str) throws Exception {
                        Log.e(TAG, "concat accept: " + str);
                    }
                });
    }


    private void flatMap() {
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
            }
        }).flatMap(new Function<Integer, ObservableSource<String>>() {
            @Override
            public ObservableSource<String> apply(Integer integer) throws Exception {
                List<String> list = new ArrayList<>();
                for (int i = 0; i < 3; i++) {
                    list.add("value is " + integer);
                }
                long delay = (long) (1 + Math.random() * 10);
                return Observable.fromIterable(list).delay(delay, TimeUnit.MILLISECONDS);
            }
        }).subscribeOn(Schedulers.newThread())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<String>() {
                    @Override
                    public void accept(String s) throws Exception {
                        Log.e(TAG, "flatMap accept : " + s);
                    }
                });
    }


    private void concatMap() {
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
            }
        }).concatMap(new Function<Integer, ObservableSource<String>>() {
            @Override
            public ObservableSource<String> apply(Integer integer) throws Exception {
                List<String> list = new ArrayList<>();
                for (int i = 0; i < 3; i++) {
                    list.add("value is " + integer);
                }
                long delay = (long) (1 + Math.random() * 10);
                return Observable.fromIterable(list).delay(delay, TimeUnit.MILLISECONDS);
            }
        }).subscribeOn(Schedulers.newThread())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<String>() {
                    @Override
                    public void accept(String s) throws Exception {
                        Log.e(TAG, "concatMap accept: " + s);
                    }
                });
    }


    private void distinct() {
        Observable.just(1, 2, 3, 4, 5, 2, 3, 1)
                .distinct()
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.e(TAG, "distinct accept : " + integer);
                    }
                });
    }


    private void filter() {
        Observable.just(1, 2, 3, -1, 20, 13, 30)
                .filter(new Predicate<Integer>() {
                    @Override
                    public boolean test(Integer integer) throws Exception {
                        return integer > 15;
                    }
                })
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.e(TAG, "filter accept : " + integer);
                    }
                });
    }


    private void buffer() {
        Observable.just(1, 2, 3, 4, 5)
                .buffer(3, 2)
                .subscribe(new Consumer<List<Integer>>() {
                    @Override
                    public void accept(List<Integer> integers) throws Exception {
                        for (Integer i : integers) {
                            Log.e(TAG, "buffer accept : " + i);
                        }
                    }
                });
    }


    private void timer() {
        Log.e(TAG, "timer start: " + sdf.format(new Date()));

        Observable.timer(2, TimeUnit.SECONDS)
                .subscribe(new Consumer<Long>() {
                    @Override
                    public void accept(Long aLong) throws Exception {
                        Log.e(TAG, "timer " + aLong + " accept at " + sdf.format(new Date()));
                    }
                });
    }

    private Disposable mIntervalDisposable;


    private void interval() {
        Log.e(TAG, "interval start: " + sdf.format(new Date()));
        mIntervalDisposable = Observable.interval(3, 2, TimeUnit.SECONDS)
                .subscribe(new Consumer<Long>() {
                    @Override
                    public void accept(Long aLong) throws Exception {
                        Log.e(TAG, "interval " + aLong + " accept  at " + sdf.format(new Date()));
                    }
                });
    }


    private void intervalRange() {
        Log.e(TAG, "interval start: " + sdf.format(new Date()));
        Observable.intervalRange(4, 3, 2, 2, TimeUnit.SECONDS)
                .subscribe(new Consumer<Long>() {
                    @Override
                    public void accept(Long aLong) throws Exception {
                        Log.e(TAG, "intervalRange " + aLong + " accept  at " + sdf.format(new Date()));

                    }
                });
    }


    private void doOnNext() {
        Observable.just(1, 2, 3, 4, 5)
                .doOnNext(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.e(TAG, "doOnNext do something like save " + integer);
                    }
                }).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.e(TAG, "doOnNext accept: " + integer);
            }
        });
    }


    private void skip() {
        Observable.just(1, 2, 3, 4, 5)
                .skip(2)
                .skipLast(2)
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.e(TAG, "skip accept: " + integer);
                    }
                });
    }


    private void take() {
        Observable.just(1, 2, 3, 4, 5)
                .take(2)
                .takeLast(1)
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.e(TAG, "take accept: " + integer);
                    }
                });
    }

    private void single() {
        Single.just(new Random().nextInt()).subscribe(new SingleObserver<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.e(TAG, "single onSubscribe");
            }

            @Override
            public void onSuccess(Integer integer) {
                Log.e(TAG, "single onSuccess: " + integer);
            }

            @Override
            public void onError(Throwable e) {
                Log.e(TAG, "single onError: " + e.getMessage());
            }
        });
    }

    private void debounce() {
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                Thread.sleep(300);
                emitter.onNext(2);
                Thread.sleep(600);
                emitter.onNext(3);
                Thread.sleep(400);
                emitter.onNext(4);
                Thread.sleep(700);
                emitter.onNext(5);
                Thread.sleep(800);
            }
        }).debounce(500, TimeUnit.MILLISECONDS)
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.e(TAG, "debounce accept: " + integer);
                    }
                });
    }

    private void defer() {
        final StringBuilder sb = new StringBuilder();
        Observable<Integer> observable = Observable.defer(new Callable<ObservableSource<Integer>>() {
            @Override
            public ObservableSource<Integer> call() throws Exception {
                sb.append(20);
                return Observable.just(1, 2, 3);
            }
        });
        Log.e(TAG, "defer before subscribe sb : " + sb.toString());

        observable.subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {

            }

            @Override
            public void onNext(Integer integer) {
                sb.append(integer);
            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onComplete() {
                Log.e(TAG, "defer after subscribe sb : " + sb.toString());
            }
        });
    }

    private void last() {
        Observable.just(1, 2, 3, 4, 5)
                .last(6)
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.e(TAG, "last accept : " + integer);
                    }
                });

        Observable.empty()
                // 如果元数据没有发送，则发送默认值
                .last(3)
                .subscribe(new Consumer<Object>() {
                    @Override
                    public void accept(Object o) throws Exception {
                        Log.e(TAG, "empty last accept : " + o);
                    }
                });
    }

    private void merge() {
        Observable.merge(Observable.just(1, 2, 3).delay(200, TimeUnit.MILLISECONDS), Observable.just(4, 5, 6), Observable.just(7, 8, 9))
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.e(TAG, "merge accept: " + integer);
                    }
                });
    }

    private void reduce() {
        Observable.just(1, 2, 3)
                .reduce(new BiFunction<Integer, Integer, Integer>() {
                    @Override
                    public Integer apply(Integer integer, Integer integer2) throws Exception {
                        return integer * integer + integer2; // (1 * 1 + 2) * (1 * 1 + 2) + 3
                    }
                })
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.e(TAG, "reduce accept : " + integer);
                    }
                });
    }

    private void scan() {
        Observable.just(1, 2, 3)
                .scan(5, new BiFunction<Integer, Integer, Integer>() {
                    @Override
                    public Integer apply(Integer integer, Integer integer2) throws Exception {
                        return integer * integer + integer2;
                    }
                })
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.e(TAG, "scan accept : " + integer);
                    }
                });
    }

    private void window() {
        Observable.just(1, 2, 3, 4, 5, 6, 7, 8)
                .window(3)
                .subscribe(new Consumer<Observable<Integer>>() {
                    @Override
                    public void accept(Observable<Integer> integerObservable) throws Exception {
                        Log.e(TAG, "window accept begin...");
                        integerObservable.subscribe(new Consumer<Integer>() {
                            @Override
                            public void accept(Integer integer) throws Exception {
                                Log.e(TAG, "window accept : " + integer);
                            }
                        });
                    }
                });

        Observable.interval(1, TimeUnit.SECONDS)
                .take(15)
                .window(3, TimeUnit.SECONDS)
                .subscribe(new Consumer<Observable<Long>>() {
                    @Override
                    public void accept(Observable<Long> longObservable) throws Exception {
                        Log.e(TAG, "time window accept begin...");
                        longObservable.subscribe(new Consumer<Long>() {
                            @Override
                            public void accept(Long aLong) throws Exception {
                                Log.e(TAG, "time window accept : " + aLong);
                            }
                        });
                    }
                });
    }

}
